package com.snda.dw.lidejia.hdfsoverftp;

import java.io.BufferedInputStream;
import java.io.File;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

import org.apache.ftpserver.ConnectionConfigFactory;
import org.apache.ftpserver.DataConnectionConfigurationFactory;
import org.apache.ftpserver.FtpServer;
import org.apache.ftpserver.FtpServerFactory;
import org.apache.ftpserver.ftplet.UserManager;
import org.apache.ftpserver.listener.Listener;
import org.apache.ftpserver.listener.ListenerFactory;
import org.apache.ftpserver.usermanager.ClearTextPasswordEncryptor;
import org.apache.ftpserver.usermanager.PropertiesUserManagerFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class HdfsOverFtpServer {

    private static final Logger LOGGER = LoggerFactory.getLogger(HdfsOverFtpServer.class);
    //
    //private static final String USERS_CONF = "users.conf";
    //private static final String CONF_FILE = "hdfs-over-ftp.conf";
    private static int port = 2222;
    private static int sslPort = 2226;
    //
    private static String passivePorts = null;
    private static String sslPassivePorts = null;
    private static String hdfsUri = null;
    private static String sslPassword = null;
    //
    private static int threadNum = 30;

    public static void main(String[] args) throws Exception {
        loadConfig();

        startServer();

        //startSSLServer();
    }

    private static void loadConfig() throws IOException {
        BufferedInputStream propertiesStream = new BufferedInputStream(HdfsOverFtpServer.class.getResourceAsStream("/hdfs-over-ftp.conf"));
        try {
            Properties props = new Properties();
            props.load(propertiesStream);
            //
            try {
                port = Integer.parseInt(props.getProperty("port"));
                LOGGER.info(String.format("Port: %s", port));
            } catch (Exception e) {
                LOGGER.error("port is not set");
                System.exit(-1);
            }

            passivePorts = props.getProperty("data-ports");
            LOGGER.info(String.format("Data ports: %s", passivePorts));
            if (passivePorts == null) {
                LOGGER.error("data-ports is not set");
                System.exit(-1);
            }
//            try {
//                sslPort = Integer.parseInt(props.getProperty("ssl-port"));
//                LOGGER.info(String.format("ssl-port is %s.", sslPort));
//            } catch (Exception e) {
//                LOGGER.info("ssl-port is not set. so ssl server will not be started");
//            }
//
//            sslPassivePorts = props.getProperty("ssl-data-ports");
//            if (sslPassivePorts == null) {
//                LOGGER.error("ssl-data-ports is not set");
//                System.exit(1);
//            }
//            sslPassword = props.getProperty("keystore-password");
//            if (sslPassword == null) {
//                LOGGER.error("keystore-password is not set");
//                System.exit(1);
//            }

            hdfsUri = props.getProperty("hdfs-uri");
            if (hdfsUri == null) {
                LOGGER.error("hdfs-uri is not set");
                System.exit(-1);
            } else {
                HdfsOverFtpSystem.setHDFS_URI(hdfsUri);
                LOGGER.info(String.format("HDFS URI: %s", hdfsUri));
            }

            try {
                threadNum = Integer.parseInt(props.getProperty("max-threads"));
            } catch (Exception e) {
                threadNum = 100;
            }
            LOGGER.info(String.format("Max thread num is %s", threadNum));

            String superuser = props.getProperty("superuser");
            if (superuser == null) {
                LOGGER.error("superuser is not set");
                System.exit(1);
            } else {
                LOGGER.info(String.format("Superuser is %s", superuser));
                HdfsOverFtpSystem.setSuperuser(superuser);
            }
        } finally {
            propertiesStream.close();
        }
    }

    public static void startServer()
            throws Exception {
        LOGGER.info("Starting Hdfs-Over-Ftp Server.");

        FtpServerFactory ftpServerFactory = new FtpServerFactory();

        // Data connection configuration
        DataConnectionConfigurationFactory dataConFactory = new DataConnectionConfigurationFactory();
        dataConFactory.setPassivePorts(passivePorts);

        // Listener
        ListenerFactory listenerFactory = new ListenerFactory();
        listenerFactory.setDataConnectionConfiguration(dataConFactory.createDataConnectionConfiguration());
        listenerFactory.setPort(port);
        Map<String, Listener> listenerMap = new HashMap<String, Listener>();
        listenerMap.put("default", listenerFactory.createListener());

        // User manager
        PropertiesUserManagerFactory userManagerFactory = new PropertiesUserManagerFactory();
        userManagerFactory.setFile(new File(HdfsOverFtpServer.class.getResource("/users.conf").toURI()));
        userManagerFactory.setPasswordEncryptor(new ClearTextPasswordEncryptor());
        UserManager um = userManagerFactory.createUserManager();

        // Connection configuration
        ConnectionConfigFactory connectionConfigFactory = new ConnectionConfigFactory();
        connectionConfigFactory.setMaxLogins(threadNum);
        //
        ftpServerFactory.setConnectionConfig(connectionConfigFactory.createConnectionConfig());
        ftpServerFactory.setListeners(listenerMap);
        ftpServerFactory.setUserManager(um);
        ftpServerFactory.setFileSystem(new HdfsFileSystemFactory());
        //
        FtpServer ftpServer = ftpServerFactory.createServer();
        ftpServer.start();
    }
//    public static void startSSLServer()
//            throws Exception {
//        LOGGER.info("Starting Hdfs-Over-Ftp SSL Server.");
//        LOGGER.info(String.format("SLL port: %s", sslPort));
//        LOGGER.info(String.format("SSL-data-ports: %s", sslPassivePorts));
//        LOGGER.info(String.format("HDFS-URI: %s", hdfsUri));
//
//        HdfsOverFtpSystem.setHDFS_URI(hdfsUri);
//
//        FtpServerFactory ftpServerFactory = new FtpServerFactory();
//
//        // Data connection configuration
//        DataConnectionConfigurationFactory dataConFactory = new DataConnectionConfigurationFactory();
//        dataConFactory.setPassivePorts(sslPassivePorts);
//
//        // SSL configuration
//        SslConfigurationFactory sslConfigurationFactory = new SslConfigurationFactory();
//        sslConfigurationFactory.setKeystoreFile(new File("ftp.jks"));
//        sslConfigurationFactory.setKeystoreType("JKS");
//        sslConfigurationFactory.setKeystorePassword(sslPassword);
//
//        // Listener
//        ListenerFactory listenerFactory = new ListenerFactory();
//        listenerFactory.setDataConnectionConfiguration(dataConFactory.createDataConnectionConfiguration());
//        listenerFactory.setPort(sslPort);
//        listenerFactory.setSslConfiguration(sslConfigurationFactory.createSslConfiguration());
//        listenerFactory.setImplicitSsl(true);
//        Map<String, Listener> listenerMap = new HashMap<String, Listener>();
//        listenerMap.put("default", listenerFactory.createListener());
//
//        // User manager
//        PropertiesUserManagerFactory userManagerFactory = new PropertiesUserManagerFactory();
//        userManagerFactory.setFile(new File("users.conf"));
//        userManagerFactory.setPasswordEncryptor(new ClearTextPasswordEncryptor());
//        UserManager um = userManagerFactory.createUserManager();
//        //
//        String[] userNames = um.getAllUserNames();
//        StringBuilder sb = new StringBuilder();
//        for (int i = 0; i < userNames.length; i++) {
//            sb.append(userNames[i]);
//            sb.append(",");
//        }
//        LOGGER.info(String.format("User is ", sb.toString()));
//        LOGGER.info(String.format("User mmhadoop's home directory is ", um.getUserByName("mmhadoop").getHomeDirectory()));
//
//        // Connection configuration
//        ConnectionConfigFactory connectionConfigFactory = new ConnectionConfigFactory();
//        connectionConfigFactory.setMaxLogins(1000);
//
//        ftpServerFactory.setConnectionConfig(connectionConfigFactory.createConnectionConfig());
//        ftpServerFactory.setListeners(listenerMap);
//        ftpServerFactory.setUserManager(um);
//        ftpServerFactory.setFileSystem(new HdfsFileSystemFactory());
//
//        FtpServer ftpServer = ftpServerFactory.createServer();
//        ftpServer.start();
//    }
}